package cn.lsh.spark.sql;

import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.util.*;

/**
 * 网站pr计算公式：pr = (1-d)/n + d*sum(tr)
 * d：阻尼系数，为0.85
 * n：网站总数，这里即是记录条数
 * tr：网站得到其他网站的投票权值
 */
public class PageRankSpark {
	/**
	 * 页面pr值差值范围，用于判断是否停止迭代
	 */
	private static final double LIMIT = 0.001;

	/**
	 * 阻尼系数d
	 */
	private static final double D = 0.85;

	/**
	 * 网页初始pr为1
	 */
	private static final double INIT_PR = 1;

	public static void main(String[] args) {
		SparkSession ss = SparkSession.builder().master("local")
				.appName("page_rank")
				.getOrCreate();
		JavaSparkContext jsc = new JavaSparkContext(ss.sparkContext());
		/*@step1 加载网页数据*/
		JavaRDD<String> textRdd = jsc.textFile("file:/bigdata/hadoop-test/input/pagerank/page_data.txt");
		textRdd.cache();
		//网页对应pr值的map
		Map<String, Double> pagePr = new HashMap<>();
		/*@step2 设置网页的初始化pr值*/
		textRdd.map(new Function<String, String>() {
			private static final long serialVersionUID = 964971064235782420L;

			@Override
			public String call(String v1) throws Exception {
				return v1.split(" ")[0];
			}
		}).collect().forEach(p -> pagePr.put(p, INIT_PR));
		/*@step3 广播网页的初始化pr值map*/
		Broadcast<Map<String, Double>> broadcast = jsc.broadcast(pagePr);
		//迭代计算
		while (true) {
			/*@step4 计算各网页的tr值*/
			JavaPairRDD<String, Double> pageGetTrRdd = textRdd.flatMapToPair(new PairFlatMapFunction<String, String, Double>() {
				private static final long serialVersionUID = -3369519627121102880L;

				@Override
				public Iterator<Tuple2<String, Double>> call(String s) throws Exception {
					List<Tuple2<String, Double>> ret = new ArrayList<>();
					String[] pages = s.split(" ");
					for (int i = 1; i < pages.length; i++) {
						ret.add(new Tuple2<>(pages[i], broadcast.value().get(pages[0]) / (pages.length - 1)));
					}
					return ret.iterator();
				}
			}).reduceByKey(Double::sum);
			List<Tuple2<String, Double>> collect = pageGetTrRdd.collect();
			int jTotal = 0;
			for (Tuple2<String, Double> t : collect) {
				/*@step5 重新计算网页的pr值*/
				double newPr = (1 - D) / pagePr.size() + D * t._2;
				System.out.println(t._1 + " - " + newPr);

				/*@step6 计算新旧pr值的差值*/
				double c = newPr - pagePr.get(t._1);
				int j = Math.abs((int) (c * 1000));
				jTotal += j;
				/*@step7 重新设置网页的pr值*/
				pagePr.put(t._1, newPr);
			}
			System.out.println("pr差值总和：" + jTotal);
			double avgd = jTotal / 4000.0;
			if (avgd < LIMIT) {
				/*@step8 如果符合设置的差值范围要求，退出迭代计算*/
				break;
			}
		}
	}
}
